import java.lang

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.gmall.common.Constant
import com.shellyan.gmall.rt.bean.StartupLog
import com.shellyan.gmall.rt.util.{MyKafkaUtil, RedisUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis
/**
 * @author Shelly An
 * @create 2020/9/7 16:18
 */
object DauAppOld {
  def main(args: Array[String]): Unit = {
    //1. 先创建StreamingContext
    val conf = new SparkConf().setMaster("local[2]").setAppName("DauApp")
    val ssc = new StreamingContext(conf, Seconds(3))
    //2. 从kafka得到一个stream
    val sourceStream =
      MyKafkaUtil.getKafkaStream(ssc, "DauApp", Set(Constant.STARTUP_TOPIC))
        .map(json => {
          JSON.parseObject(json, classOf[StartupLog])
        })

    sourceStream.print()
    //3. 各种转换 map,mapPartitions,filter,...     transform后可以使用rdd的方法，更丰富
    /**
     * todo 过滤
     * 1. 只保留每个设备每天第一次启动记录
     *
     */
    /* //一条数据写一次，redis疯了，优化：一个分区写一次  （一般按照这个）
       // 再优化：就写一次 executor汇总到driver写一次（数据量不是很大）
      val filterStartupLogStream: DStream[StartupLog] = sourceStream.filter(startupLog => {
      print("过滤前："+startupLog.mid)
      // 把启动日志的mid_id写入到redis，如果返回值是1，则表示是第一次，不是1就不是第一次
      val client: Jedis = RedisUtil.getClient
      //一天一个set，只存mid，每个mid最多20个字节，日活1个亿，
      // 20* 10^8 = 2个G,一个单机的redis都没问题，企业中一般64G、128G，
      val r = client.sadd(s"dau:uids:${startupLog.logDate}", startupLog.mid)
      //写完了一定要关闭
      client.close()
      //1是成功几个的意思
      r == 1
    })*/

    //一个分区写一次  （一般按照这个） 和外界沟通的时候，需要客户端，要一个分区创建一个客户端
    val filterStartupLogStream: DStream[StartupLog] = sourceStream.mapPartitions((startupLogIt: Iterator[StartupLog]) => {
      val client: Jedis = RedisUtil.getClient
      val result: Iterator[StartupLog] = startupLogIt.filter(startupLog => {
        //print("过滤前："+startupLog.mid)
        val r: lang.Long = client.sadd(s"dau:uids:${startupLog.logDate}", startupLog.mid)
        //每次都加数据留存时间（24h）  也可如存在就不加了
        client.expire(s"dau:uids:${startupLog.logDate}",24*60*60)
        r==1
      })
      client.close()
      result
    })
    //    //实现三：流的操作转为rdd的操作  打印线程名字试一下
    //    //代码1：driver 只执行一次
    //    //集合
    //    val filterStartupLogStream: DStream[StartupLog] = sourceStream.transform(rdd => {
    //      //代码2：driver 每个批次执行一次
    //      //集合
    //      rdd.mapPartitions((startupLogIt: Iterator[StartupLog]) => {
    //        //代码3：executor 每个批次执行一次
    //        //集合要可序列化
    //        val client: Jedis = RedisUtil.getClient
    //        val result: Iterator[StartupLog] = startupLogIt.filter(startupLog => {
    //          client.sadd(s"dau:uids:${startupLog.logDate}", startupLog.mid) == 1
    //        })
    //        client.close()
    //        result
    //      })
    //    })
    //4. 输出(print foreachRDD--对外输出最好用这个)
    filterStartupLogStream.foreachRDD(rdd=>{
      //rdd就是三秒内所有的数据，不管多少分区，都在一个rdd中
      import org.apache.phoenix.spark._   //隐式转换，使rdd可以使用pheonix包中的rdd方法
      rdd.saveToPhoenix("GMALL_DAU",
        //从rdd中到seq中，按顺序，从seq到表里，按名称
        Seq("MID", "UID", "APPID", "AREA", "OS", "CHANNEL", "LOGTYPE", "VERSION", "TS", "LOGDATE", "LOGHOUR"),
        zkUrl = Option("hadoop102,hadoop103,hadoop104:2181")
      )
    })
    //5. 启动上下文
    ssc.start()
    //6. 阻塞
    ssc.awaitTermination()

  }
}
